package com.lagou.manager;

import com.lagou.client.RPCConsumer;
import com.lagou.service.IUserService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryNTimes;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
public class ZkManager implements InitializingBean, DisposableBean {
    private CuratorFramework client;

    public Map<Integer, IUserService> serviceMap = new ConcurrentHashMap<>();

    @Override
    public void afterPropertiesSet() throws Exception {
        RetryNTimes retryNTimes = new RetryNTimes(0, 1000);
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", retryNTimes);
        client.start();

        CuratorCache watcher = CuratorCache.builder(client, "/providers")
                .build();
        watcher.listenable()
                .addListener(CuratorCacheListener.builder()
                        .forPathChildrenCache("/providers", client,
                                (client1, event) -> {
                                    System.out.println(event.getType().name());
                                    if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                                        String path = event.getData().getPath();
                                        String[] split = path.split("/");
                                        int intPort = Integer.parseInt(split[split.length - 1]);
                                        IUserService service = (IUserService) RPCConsumer.createProxy(IUserService.class, intPort);
                                        serviceMap.put(intPort, service);
                                    }
                                    if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                                        String path = event.getData().getPath();
                                        String[] split = path.split("/");
                                        int intPort = Integer.parseInt(split[split.length - 1]);
                                        System.out.println(intPort);
                                        serviceMap.remove(intPort);
                                        RPCConsumer.eventLoopGroupHashMap.get(intPort).shutdownGracefully();
                                    }
                                })
                        .build());
        watcher.start();
    }

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    @Override
    public void destroy() throws Exception {
        scheduledThreadPoolExecutor.shutdown();
        boolean b = scheduledThreadPoolExecutor.awaitTermination(10L, TimeUnit.SECONDS);
        if (!b) {
            scheduledThreadPoolExecutor.shutdownNow();
        }
        client.close();
    }
}
